package com.xxxx;

import com.xxxx.pojo.Person;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

public class Hello01FlinkStreamTable {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        //获取数据源
        DataStreamSource<String> streamSource = environment.fromCollection(Arrays.asList(
                "1,xiaoming,18,91,1",
                "2,xiaoli,20,90,1",
                "3,xiaodong,18,92,2",
                "4,xiaogang,21,78,2",
                "5,xiaohong,16,100,1"
        ));
        //输入数据的时候必须按照上面的格式进行操作
        //DataStreamSource<String> streamSource = env.socketTextStream("192.168.201.101", 9999);

        //将流式的一行数据转成一个Person对象
        SingleOutputStreamOperator<Person> personStream = streamSource.map(new MapFunction<String, Person>() {
            @Override
            public Person map(String value) throws Exception {
                String[] split = value.split(",");
                return new Person(split[0], split[1], Integer.parseInt(split[2]), Integer.valueOf(split[3]), split[4]);
            }
        });
        //将Stream流转成Table
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
        tableEnvironment.registerDataStream("t_person", personStream);
        //执行SQL
        Table table = tableEnvironment.sqlQuery("select name,score from t_person where score = (select max(score) from t_person)");
        //tableEnvironment.scan("").where().groupBy().select().orderBy()

        //Table table = tableEnvironment.fromDataStream(personStream);
        //Table table1 = table.as("a,b,c,d,e");

        //打印表的结构
        table.printSchema();
        DataStream<Tuple2<Boolean, Row>> stream = tableEnvironment.toRetractStream(table, Row.class);
        stream.print();
        //运行环境
        environment.execute("Hello01FlinkStreamTable" + System.currentTimeMillis());
    }
}
